package com.swak.consistency.tcc.storage.impls;

import java.time.LocalDateTime;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

import javax.sql.DataSource;

import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;

import com.swak.App;
import com.swak.consistency.core.execption.ConsistencyException;
import com.swak.consistency.tcc.context.TransLog;
import com.swak.consistency.tcc.context.TransParticipant;
import com.swak.consistency.tcc.storage.TransLogStorage;
import com.swak.persistence.DatabaseStorage;
import com.swak.utils.CollectionUtils;
import com.swak.utils.Ints;

/**
 * 事务日志服务
 * 
 * @author DELL
 */
public class TransLogStorageForMysql extends DatabaseStorage implements TransLogStorage, InitializingBean, DisposableBean {

	private String module;
	private String tableName;
	private String failTableName;

	public TransLogStorageForMysql(DataSource dataSource, String module) {
		super(dataSource);
		this.module = module;
		this.tableName = "tcc_" + this.module + "_log";
		this.failTableName = "tcc_" + this.module + "_fail";
	}

	@Override
	public void afterPropertiesSet() throws Exception {
		this.createTable();
	}

	private void createTable() {

		// 事务日志表
		String buildTableSQL = "CREATE TABLE IF NOT EXISTS `" + tableName + "` (" + "  `trans_id` BIGINT NOT NULL,"
				+ "  `phase` tinyint NOT NULL," + "  `role` tinyint NOT NULL," + "  `retried_times` tinyint NOT NULL,"
				+ "  `version` tinyint NOT NULL," + "  `target_class` varchar(256) ,"
				+ "  `target_method` varchar(128) ," + "  `invocation` longblob," + "  `create_time` datetime NOT NULL,"
				+ "  `update_time` datetime NOT NULL,"
				+ "  PRIMARY KEY (`trans_id`)) ENGINE=INNODB DEFAULT CHARSET=UTF8 COMMENT='分布式事务'";

		this.executeUpdate(buildTableSQL);

		// 重试失败的事务日志表
		buildTableSQL = "CREATE TABLE IF NOT EXISTS `" + failTableName + "` (" + "  `trans_id` BIGINT NOT NULL,"
				+ "  `phase` tinyint NOT NULL," + "  `role` tinyint NOT NULL," + "  `retried_times` tinyint NOT NULL,"
				+ "  `version` tinyint NOT NULL," + "  `target_class` varchar(256) ,"
				+ "  `target_method` varchar(128) ," + "  `invocation` longblob," + "  `create_time` datetime NOT NULL,"
				+ "  `update_time` datetime NOT NULL,"
				+ "  PRIMARY KEY (`trans_id`)) ENGINE=INNODB DEFAULT CHARSET=UTF8 COMMENT='分布式事务  - 失败'";

		this.executeUpdate(buildTableSQL);
	}

	/**
	 * 查询当前事务
	 * 
	 * @param transId
	 * @return
	 */
	@Override
	public TransLog get(Long transId) {
		String selectSql = "select * from " + tableName + " where trans_id = ? ";
		List<Map<String, Object>> list = executeQuery(selectSql, transId);
		if (list != null && list.size() > 0) {
			return list.stream().filter(Objects::nonNull).map(this::buildResultMap).findFirst().orElse(null);
		}
		return null;
	}

	/**
	 * 执行保存
	 * 
	 * @param transLog
	 * @return
	 */
	@Override
	public void create(TransLog transLog) {
		String sql = "INSERT INTO " + tableName
				+ "(trans_id,phase,role,retried_times,version,target_class,target_method,create_time,update_time,invocation)"
				+ " VALUES(?,?,?,?,?,?,?,?,?,?)";
		try {
			final byte[] serialize = App.serialize(transLog.getParticipants());
			executeUpdate(sql, transLog.getTransId(), transLog.getPhase(), transLog.getRole(), transLog.getRetryTimes(),
					transLog.getVersion(), transLog.getTargetClass(), transLog.getTargetMethod(),
					transLog.getCreateTime(), transLog.getUpdateTime(), serialize);
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
	}

	/**
	 * 修改事务所处的阶段
	 * 
	 * @param transId
	 * @param phase
	 * @return
	 */
	@Override
	public void updatePhase(TransLog trans) {
		try {
			String sql = "update " + tableName + " set phase = ? , update_time = ?  where trans_id = ?  ";
			executeUpdate(sql, trans.getPhase(), new Date(), trans.getTransId());
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
	}

	/**
	 * 修改事务的参与者
	 * 
	 * @param transLog
	 */
	@Override
	public void updateParticipants(TransLog transLog) {
		try {
			final byte[] serialize = App.serialize(transLog.getParticipants());
			String sql = "update " + tableName + " set invocation = ? , update_time = ? where trans_id = ? ";
			executeUpdate(sql, serialize, new Date(), transLog.getTransId());
		} catch (Exception e) {
			e.printStackTrace();
			throw new ConsistencyException(e);
		}
	}

	/**
	 * 删除事务日志（READY阶段的事务日志不能删除，通过定时任务去删除）
	 * 
	 * @param transId
	 * @return
	 */
	@Override
	public void remove(TransLog trans) {
		try {
			String sql = "delete from " + tableName + " where trans_id = ? and phase != 0 ";
			executeUpdate(sql, trans.getTransId());
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
	}

	@SuppressWarnings("unchecked")
	private TransLog buildResultMap(final Map<String, Object> map) {
		TransLog transLog = new TransLog();
		transLog.setTransId((Long) map.get("trans_id"));
		transLog.setPhase((Integer) map.get("phase"));
		transLog.setRole((Integer) map.get("role"));
		transLog.setRetryTimes((Integer) map.get("retried_times"));
		transLog.setVersion((Integer) map.get("version"));
		transLog.setCreateTime((LocalDateTime) map.get("create_time"));
		transLog.setUpdateTime((LocalDateTime) map.get("update_time"));
		byte[] bytes = (byte[]) map.get("invocation");
		try {
			final List<TransParticipant> miloParticipantBeans = (List<TransParticipant>) App.deserialize(bytes);
			transLog.setParticipants(miloParticipantBeans);
		} catch (Exception e) {
			throw new ConsistencyException(e);
		}
		return transLog;
	}

	/**
	 * 失败的事务日志
	 */
	@Override
	public List<TransLog> queryAllByDelay(Date delay) {
		String selectSql = "select * from " + tableName + " where update_time < ? ";
		List<Map<String, Object>> list = executeQuery(selectSql, delay);
		if (CollectionUtils.isNotEmpty(list)) {
			return list.stream().filter(Objects::nonNull).map(this::buildResultMap).collect(Collectors.toList());
		}
		return Collections.emptyList();
	}

	@Override
	public void retry(TransLog transLog, int maxTimes) {
		transLog.setRetryTimes(Ints.addI(transLog.getRetryTimes(), 1));
		if (transLog.getRetryTimes() > maxTimes) {
			this.executeMultiSql((conn) -> {
				String sql = "INSERT INTO " + failTableName
						+ "(trans_id,phase,role,retried_times,version,target_class,target_method,create_time,update_time,invocation)"
						+ " VALUES(?,?,?,?,?,?,?,?,?,?)";
				try {
					final byte[] serialize = App.serialize(transLog.getParticipants());
					executeMultiSql_Update(conn, sql, transLog.getTransId(), transLog.getPhase(), transLog.getRole(),
							transLog.getRetryTimes(), transLog.getVersion(), transLog.getTargetClass(),
							transLog.getTargetMethod(), transLog.getCreateTime(), transLog.getUpdateTime(), serialize);
				} catch (Exception e) {
					throw new ConsistencyException(e);
				}

				sql = "delete from " + tableName + " where trans_id = ?";
				executeMultiSql_Update(conn, sql, transLog.getTransId());
			});
		} else {
			String sql = "update " + tableName + " set retried_times = ? , update_time = ? where trans_id = ?";
			executeUpdate(sql, transLog.getRetryTimes(), new Date(), transLog.getTransId());
		}
	}

	@Override
	public void destroy() throws Exception {
		try {

		} catch (Exception e) {
		}
	}
}